// 似乎只有用类封装才能用，不然套接字、缓冲区根本无法保存
// g++ testasio.cpp -lpthread
// g++ testasio.cpp -lpthread -lssl -lcrypto

#include <iostream>
#include <asio.hpp>

class tcp_session : public std::enable_shared_from_this<tcp_session>
{
public:
	tcp_session(asio::ip::tcp::socket sock)
		: _sock(std::move(sock))
	{
	}

public:
	void start()
	{
		auto self(shared_from_this());

		_recv();
	}

private:
	void _recv()
	{
		auto self(shared_from_this());

		_sock.async_receive(asio::buffer(_buf, _buffer_size), 0, std::bind(&tcp_session::_recv_handler, self, std::placeholders::_1, std::placeholders::_2));
	}

	void _recv_handler(asio::error_code ec, std::size_t trans)
	{
		// auto self(shared_from_this());

		if (ec)
		{
			std::cout << "recv error: " << ec << std::endl;
			return;
		}

		std::cout.write(_buf, trans);
		std::cout << std::endl;

		_recv();
	}

private:
	enum
	{
		_buffer_size = 256
	};
	char _buf[_buffer_size];
	asio::ip::tcp::socket _sock;
};

class tcp_server
{
public:
	tcp_server(asio::io_context &context, unsigned short port, const std::string &ip = "0.0.0.0")
		: _acceptor(context, asio::ip::tcp::endpoint(asio::ip::make_address(ip), port))
	{
		_accept();
	}

private:
	void _accept()
	{
		_acceptor.async_accept(std::bind(&tcp_server::_accept_handler, this, std::placeholders::_1, std::placeholders::_2));
	}

	void _accept_handler(asio::error_code ec, asio::ip::tcp::socket sock)
	{
		if (!ec)
		{
			std::cout << sock.remote_endpoint().address().to_string() << ":" << sock.remote_endpoint().port() << " connected" << std::endl;

			std::make_shared<tcp_session>(std::move(sock))->start();
		}

		_accept();
	}

private:
	asio::ip::tcp::acceptor _acceptor;
};

class tcp_client
{
	asio::ip::tcp::socket _client;
	std::string _host;
	std::string _port;
	asio::ip::tcp::resolver _resolver;

	enum
	{
		_buffer_size = 256
	};
	char _buf[_buffer_size];

	asio::deadline_timer _timer;

public:
	tcp_client(asio::io_context &context, const std::string &host, const std::string &port)
		: _client(context), _host(host), _port(port), _resolver(context), _timer(context)
	{
		_connect();
	}

private:
	void _connect()
	{
		_resolver.async_resolve(asio::ip::tcp::v4(), _host, _port, [this](asio::error_code ec, asio::ip::tcp::resolver::results_type results)
								{
									if (ec)
										return;

										auto ep = results.cbegin()->endpoint();

										std::cout << ep.address() << ":" << ep.port() << std::endl;

										_client.async_connect(ep, std::bind(&tcp_client::_connect_handler, this, std::placeholders::_1)); });
	}

	void _connect_handler(asio::error_code ec)
	{
		if (ec)
		{
			std::cout << "connect error: " << ec << std::endl;

			_reconnect();
			return;
		}

		std::cout << "connected to " << _host << ":" << _port << std::endl;

		_recv();
	}

	void _reconnect()
	{
		_client.close();

		_timer.expires_from_now(boost::posix_time::seconds(3));
		_timer.async_wait([this](const asio::error_code &ec)
						  { _connect(); });
	}

	void _recv()
	{
		_client.async_receive(asio::buffer(_buf, _buffer_size), 0, std::bind(&tcp_client::_recv_handler, this, std::placeholders::_1, std::placeholders::_2));
	}

	void _recv_handler(asio::error_code ec, std::size_t trans)
	{
		if (ec)
		{
			std::cout << "recv error: " << ec << std::endl;

			_reconnect();
			return;
		}

		std::cout.write(_buf, trans);
		std::cout << std::endl;

		_recv();
	}
};

class udp_server
{
	asio::ip::udp::socket _sock;

	std::string _host;
	std::string _port;

	enum
	{
		_buffer_size = 256
	};
	char _buf[_buffer_size];

	asio::ip::udp::endpoint _recv_ep;

public:
	udp_server(asio::io_context &context, int port, const std::string &ip = "0.0.0.0")
		: _sock(context, asio::ip::udp::endpoint(asio::ip::address::from_string(ip), port))
	{
		_recv();
	}

private:
	void _recv()
	{
		_sock.async_receive_from(asio::buffer(_buf, _buffer_size), _recv_ep, std::bind(&udp_server::_recv_handler, this, std::placeholders::_1, std::placeholders::_2));
	}

	void _recv_handler(asio::error_code ec, std::size_t trans)
	{
		if (ec)
		{
			std::cout << "recv error: " << ec << std::endl;
			return;
		}

		std::cout << _recv_ep << ": ";
		std::cout.write(_buf, trans);
		std::cout << std::endl;

		_recv();
	}
};

class serialport
{
	asio::serial_port _serial;

	int _baud_rate;
	int _data_len;
	asio::serial_port_base::stop_bits::type _stop_bit;
	asio::serial_port_base::parity::type _parity;

	enum
	{
		_buffer_size = 256
	};
	char _buf[_buffer_size];

public:
	serialport(asio::io_context &context, const std::string &dev_name, int baud_rate, int data_len, asio::serial_port_base::stop_bits::type stop_bit, asio::serial_port_base::parity::type parity)
		: _serial(context, dev_name), _baud_rate(baud_rate), _data_len(data_len), _stop_bit(stop_bit), _parity(parity)
	{
		_open();
	}

public:
	size_t write(const char *data, size_t len)
	{
		return _serial.write_some(asio::buffer(data, len));
	}

	bool opened()
	{
		return _serial.is_open();
	}

private:
	void _open()
	{
		_serial.set_option(asio::serial_port_base::baud_rate(_baud_rate));
		_serial.set_option(asio::serial_port_base::character_size(_data_len));
		_serial.set_option(asio::serial_port_base::stop_bits(_stop_bit));
		_serial.set_option(asio::serial_port_base::parity(_parity));

		_recv();
	}

	void _recv()
	{
		_serial.async_read_some(asio::buffer(_buf, _buffer_size), std::bind(&serialport::_recv_handler, this, std::placeholders::_1, std::placeholders::_2));
	}

	void _recv_handler(asio::error_code ec, std::size_t trans)
	{
		if (ec)
		{
			std::cout << "recv error: " << ec << std::endl;
			return;
		}

		std::cout.write(_buf, trans);
		std::cout << std::endl;

		_recv();
	}
};

void test_serial(asio::io_context &context)
{
	// 得先创建这两个口
	auto port1 = std::make_shared<serialport>(context, "/dev/pts/4", 9600, 8, asio::serial_port_base::stop_bits::one, asio::serial_port_base::parity::none);
	auto port2 = std::make_shared<serialport>(context, "/dev/pts/5", 9600, 8, asio::serial_port_base::stop_bits::one, asio::serial_port_base::parity::none);

	std::thread([&context, port1, port2]()
				{
		while(!context.stopped())
		{			
port1->write("port1", 5);
port2->write("port2", 5);

			std::this_thread::sleep_for(std::chrono::seconds(1));
		} })
		.detach();
}

#include <asio/ssl.hpp>

class ssl_client
{
	asio::ssl::stream<asio::ip::tcp::socket> _client;
	std::string _host;
	std::string _port;
	asio::ip::tcp::resolver _resolver;

	enum
	{
		_buffer_size = 256
	};
	char _buf[_buffer_size];

	asio::deadline_timer _timer;
	bool _connected;

public:
	ssl_client(asio::io_context &context, asio::ssl::context &scontext, const std::string &host, const std::string &port)
		: _client(context, scontext), _host(host), _port(port), _resolver(context), _timer(context), _connected(false)
	{
		_client.set_verify_mode(asio::ssl::verify_peer);
		_client.set_verify_callback(std::bind(&ssl_client::verify_certificate, this, std::placeholders::_1, std::placeholders::_2));

		_connect();
	}

public:
	size_t write(const char *data, size_t len)
	{
		return _client.write_some(asio::buffer(data, len));
	}

	bool connected()
	{
		return _connected;
	}

private:
	void _connect()
	{
		_resolver.async_resolve(asio::ip::tcp::v4(), _host, _port, [this](asio::error_code ec, asio::ip::tcp::resolver::results_type results)
								{
									if (ec)
										return;

										auto ep = results.cbegin()->endpoint();

										std::cout << ep.address() << ":" << ep.port() << std::endl;

asio::async_connect(_client.lowest_layer(), results, std::bind(&ssl_client::_connect_handler, this, std::placeholders::_1)); });
	}

	void _connect_handler(asio::error_code ec)
	{
		if (ec)
		{
			std::cout << "connect error: " << ec.message() << std::endl;

			_reconnect();
			return;
		}

		std::cout << "connected to " << _host << ":" << _port << std::endl;

		handshake();
	}

	void handshake()
	{
		_client.async_handshake(asio::ssl::stream_base::client,
								[this](const std::error_code &error)
								{
									if (error)
									{
										std::cout << "Handshake failed: " << error.message() << "\n";
										return;
									}

									std::cout << "client handshake done" << std::endl;

									_connected = true;

									_recv();
								});
	}

	bool verify_certificate(bool preverified,
							asio::ssl::verify_context &ctx)
	{
		// The verify callback can be used to check whether the certificate that is
		// being presented is valid for the peer. For example, RFC 2818 describes
		// the steps involved in doing this for HTTPS. Consult the OpenSSL
		// documentation for more details. Note that the callback is called once
		// for each certificate in the certificate chain, starting from the root
		// certificate authority.

		// In this example we will simply print the certificate's subject name.
		char subject_name[256];
		X509 *cert = X509_STORE_CTX_get_current_cert(ctx.native_handle());
		X509_NAME_oneline(X509_get_subject_name(cert), subject_name, 256);
		std::cout << "Verifying " << subject_name << "\n";

		return preverified;
	}

	void _reconnect()
	{
		//_client.shutdown();

		_timer.expires_from_now(boost::posix_time::seconds(3));
		_timer.async_wait([this](const asio::error_code &ec)
						  { _connect(); });
	}

	void _recv()
	{
		_client.async_read_some(asio::buffer(_buf, _buffer_size), std::bind(&ssl_client::_recv_handler, this, std::placeholders::_1, std::placeholders::_2));
	}

	void _recv_handler(asio::error_code ec, std::size_t trans)
	{
		if (ec)
		{
			_connected = false;

			std::cout << "client recv error: " << ec << std::endl;

			_reconnect();
			return;
		}

		std::cout.write(_buf, trans);
		std::cout << std::endl;

		_recv();
	}
};

class ssl_session : public std::enable_shared_from_this<ssl_session>
{
public:
	ssl_session(asio::ip::tcp::socket sock, asio::ssl::context &sctx)
		: _sock(std::move(sock), sctx)
	{
	}

public:
	void start()
	{
		auto self(shared_from_this());

		_handshake();
	}

private:
	void _handshake()
	{
		auto self(shared_from_this());

		_sock.async_handshake(asio::ssl::stream_base::server,
							  [this, self](const std::error_code &error)
							  {
								  if (error)
								  {
									  std::cout << "serv handshake error: " << error.message() << std::endl;
									  return;
								  }

								  std::cout << "serv handshake done" << std::endl;

								  _recv();
							  });
	}

	void _recv()
	{
		auto self(shared_from_this());

		_sock.async_read_some(asio::buffer(_buf, _buffer_size), std::bind(&ssl_session::_recv_handler, self, std::placeholders::_1, std::placeholders::_2));
	}

	void _recv_handler(asio::error_code ec, std::size_t trans)
	{
		// auto self(shared_from_this());

		if (ec)
		{
			std::cout << "recv error: " << ec.message() << std::endl;
			return;
		}

		std::cout.write(_buf, trans);
		std::cout << std::endl;

		_recv();
	}

private:
	enum
	{
		_buffer_size = 256
	};
	char _buf[_buffer_size];
	asio::ssl::stream<asio::ip::tcp::socket> _sock;
};

class ssl_server
{
public:
	ssl_server(asio::io_context &context, std::shared_ptr<asio::ssl::context> sctx, unsigned short port, const std::string &ip = "0.0.0.0")
		: _acceptor(context, asio::ip::tcp::endpoint(asio::ip::make_address(ip), port)), _sctx(sctx)
	{
		_accept();
	}

private:
	void _accept()
	{
		_acceptor.async_accept(std::bind(&ssl_server::_accept_handler, this, std::placeholders::_1, std::placeholders::_2));
	}

	void _accept_handler(asio::error_code ec, asio::ip::tcp::socket sock)
	{
		if (ec)
		{
			// std::cout << "accept error: " << ec.message() << std::endl;
		}
		else
		{
			std::cout << sock.remote_endpoint().address().to_string() << ":" << sock.remote_endpoint().port() << " connected" << std::endl;

			std::make_shared<ssl_session>(std::move(sock), *_sctx)->start();
		}

		_accept();
	}

private:
	asio::ip::tcp::acceptor _acceptor;
	std::shared_ptr<asio::ssl::context> _sctx;
};

void test_ssl(asio::io_context &context)
{
	auto sctx = std::make_shared<asio::ssl::context>(asio::ssl::context::tlsv12_server);
	sctx->use_certificate_chain_file("cert_serv.cer");
	sctx->use_rsa_private_key_file("rsa_serv.pem", asio::ssl::context::pem);
	auto serv = std::make_shared<ssl_server>(context, sctx, 8901);

	auto sctx2 = std::make_shared<asio::ssl::context>(asio::ssl::context::tlsv12_client);
	sctx2->load_verify_file("cert_serv.cer");
	auto client = std::make_shared<ssl_client>(context, *sctx2, "localhost", "8901");

	std::thread([serv, sctx2, client, &context]()
				{
	while(!context.stopped())
	{
		if(client->connected())
		client->write("123qwe", 6);

		std::this_thread::sleep_for(std::chrono::seconds(1));
	} })
		.detach();
}

#include <asio/strand.hpp>

//简单日志宏，打印日志及附加的线程ID
#define LOG(a) std::cout << "[T" << std::this_thread::get_id() << "]" << a << "\n"
class Printer
{
public:
	Printer(asio::io_context &ioc)
		: m_strand(ioc),
		  m_timer1(ioc, boost::posix_time::seconds(1)),
		  m_timer2(ioc, boost::posix_time::seconds(1))
	{
		m_timer1.async_wait(asio::bind_executor(m_strand, std::bind(&Printer::print1, this)));
		m_timer2.async_wait(asio::bind_executor(m_strand, std::bind(&Printer::print2, this)));
		// m_timer1.async_wait(boost::bind(&Printer::print1, this));
		// m_timer2.async_wait(boost::bind(&Printer::print2, this));
	}

	void print1()
	{
		if (m_count++ < 10)
		{
			LOG("print 1 ...count " << m_count);
			m_timer1.expires_at(m_timer1.expires_at() + boost::posix_time::seconds(1));
			m_timer1.async_wait(asio::bind_executor(m_strand, std::bind(&Printer::print1, this)));
			// m_timer1.async_wait(boost::bind(&Printer::print1, this));
		}
	}

	void print2()
	{
		if (m_count++ < 10)
		{
			LOG("print 2 ...count " << m_count);
			m_timer2.expires_at(m_timer2.expires_at() + boost::posix_time::seconds(1));
			m_timer2.async_wait(asio::bind_executor(m_strand, std::bind(&Printer::print2, this)));
			// m_timer2.async_wait(boost::bind(&Printer::print2, this));
		}
	}

private:
	asio::io_context::strand m_strand;
	asio::deadline_timer m_timer1;
	asio::deadline_timer m_timer2;
	int m_count = 0;
};

void test_strand(asio::io_context &io_context)
{
	Printer printer(io_context);

	auto beg = boost::posix_time::second_clock::local_time();
	std::thread th([&io_context]()
				   { io_context.run(); });
	// io_context.run();
	th.join();
	auto end = boost::posix_time::second_clock::local_time();
	LOG("cost " << (end - beg).total_seconds() << " (s)");
}

class sSession : public std::enable_shared_from_this<sSession>
{
	using SessionSocketPtr = std::shared_ptr<asio::ip::tcp::socket>;

public:
	explicit sSession(SessionSocketPtr sk, asio::io_context &ioc) : m_socket(std::move(sk)), m_strand(ioc) {}

public:
	void start()
	{
		asio::async_read_until(*m_socket, m_streamBuf, "\r\n",
							   // Executor绑定到strand上，这样对应的回调函数就不会出现上面说的并发情况了
							   asio::bind_executor(this->m_strand, [self = shared_from_this()](
																	   const asio::error_code &ec,
																	   std::size_t bytes_transferred)
												   {
			if (ec)
			{
				std::cout << "session error: " << ec.message()
						  << ", thread_id: "
						  << std::this_thread::get_id()
						  << std::endl;
				return;
			}

			std::cout << "Thread: "
					  << std::this_thread::get_id()
					  << ", Get User data: "
					  << std::istream(
							 &self->m_streamBuf)
							 .rdbuf()
					  << std::endl;
			self->start(); }));
	}

private:
	SessionSocketPtr m_socket;
	asio::streambuf m_streamBuf;
	asio::io_context::strand m_strand; // 每个session会有一个strand，需要串行化的都在内部处理
};

// 异步的TCP服务器
class sTcpServer : public std::enable_shared_from_this<sTcpServer>
{
public:
	explicit sTcpServer(asio::io_context &ioc, int port) : io_context(ioc),
														   m_acceptor(ioc, asio::ip::tcp::endpoint(asio::ip::tcp::v4(), port))
	{
		std::cout << "start TcpServer...\n";
	}

	sTcpServer(sTcpServer &) = delete;
	sTcpServer(sTcpServer &&) = delete;

	void start()
	{
		auto socket = std::make_shared<asio::ip::tcp::socket>(io_context);
		m_acceptor.async_accept(*socket,
								[socket, self = shared_from_this()](asio::error_code ec)
								{
									if (ec)
									{
										std::cout << "async_accept error: " << ec.message() << std::endl;
									}

									auto session = std::make_shared<sSession>(socket, self->io_context);
									session->start();
									std::cout << "thread_id: " << std::this_thread::get_id() << ", create a session\n";
									self->start();
								});
	}

private:
	asio::io_context &io_context;
	asio::ip::tcp::acceptor m_acceptor;
};

const auto N = std::thread::hardware_concurrency();
using work_guard_type = asio::executor_work_guard<asio::io_context::executor_type>;

void test_strand2(asio::io_context &io_context)
{
	std::cout << "begin asio model" << std::endl;

	work_guard_type work_guard(io_context.get_executor());
	std::vector<std::thread> threads;
	for (auto i = 0; i < N; ++i)
	{
		threads.emplace_back(std::thread([&io_context]()
										 { io_context.run(); }));
	}

	int port;
	std::cout << "input port: ";
	std::cin >> port;

	auto tcpServer = std::make_shared<sTcpServer>(io_context, port);
	// 异步启动服务器，不阻塞
	tcpServer->start();

	for (auto &t : threads)
	{
		if (t.joinable())
		{
			t.join();
		}
	}

	std::cout << "end asio model" << std::endl;
}

int main()
{
	asio::io_context context;

	asio::signal_set ss(context);
	ss.add(SIGTERM);
	ss.async_wait([&context](const asio::error_code &ec, int signal_number)
				  {
					  if (!ec)
					  {
						  std::cout << "SIGTERM" << std::endl;

						  context.stop();
					  } });

	// tcp_server serv(context, 7834);

	// tcp_client client(context, "localhost", "7835");

	// udp_server udp(context, 7836);

	// test_serial(context);

	// test_ssl(context);

	test_strand(context);
	//test_strand2(context);

	return context.run();
}
